package com.dada.cn.paimon;

import org.rocksdb.*;

import java.util.ArrayList;
import java.util.List;

public class RocksDBReader {
    public static void main(String[] args) {
        RocksDB.loadLibrary();
        try (final Options options = new Options().setCreateIfMissing(true)) {
            // 列出所有列族
            List<byte[]> columnFamilyNames = RocksDB.listColumnFamilies(options, "/opt/soft/flink/rocksdb/job_7b558d735e42a1fdf086cd72f97d3803_op_StreamingJoinOperator_72a7ad5de6d319a422288fc6cb9392ae__1_1__uuid_593aa555-c699-4f8f-a3d1-e7e07d90458a/db");
            for (byte[] name : columnFamilyNames) {
                System.out.println("Column Family: " + new String(name));
            }
        } catch (RocksDBException e) {
            e.printStackTrace();
        }

        try (final DBOptions dbOptions = new DBOptions();
             final Options options = new Options().setCreateIfMissing(true)) {

            // 定义列族描述符
            List<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
            columnFamilyDescriptors.add(new ColumnFamilyDescriptor(RocksDB.DEFAULT_COLUMN_FAMILY));
            columnFamilyDescriptors.add(new ColumnFamilyDescriptor("left-records".getBytes()));
            columnFamilyDescriptors.add(new ColumnFamilyDescriptor("right-records".getBytes()));

            // 打开列族
            List<ColumnFamilyHandle> columnFamilyHandles = new ArrayList<>();
            RocksDB db = RocksDB.open(dbOptions, "/opt/soft/flink/rocksdb/job_7b558d735e42a1fdf086cd72f97d3803_op_StreamingJoinOperator_72a7ad5de6d319a422288fc6cb9392ae__1_1__uuid_593aa555-c699-4f8f-a3d1-e7e07d90458a/db", columnFamilyDescriptors, columnFamilyHandles);

            // 获取列族句柄
            ColumnFamilyHandle leftRecordsHandle = columnFamilyHandles.get(1);
            ColumnFamilyHandle rightRecordsHandle = columnFamilyHandles.get(2);

            // 遍历所有键值对
            RocksIterator iterator = db.newIterator();
            for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
                System.out.println("Key: " + new String(iterator.key()) + ", Value: " + new String(iterator.value()));
            }
            // 读取数据
//            byte[] value1 = db.get(leftRecordsHandle, "key1".getBytes());
//            byte[] value2 = db.get(rightRecordsHandle, "key2".getBytes());
//            System.out.println("Value1: " + new String(value1));
//            System.out.println("Value2: " + new String(value2));

            // 关闭列族句柄和数据库
            for (ColumnFamilyHandle handle : columnFamilyHandles) {
                handle.close();
            }
            db.close();
        } catch (RocksDBException e) {
            e.printStackTrace();
        }

//        try (final Options options = new Options().setCreateIfMissing(false)) {
//            try (final RocksDB db = RocksDB.open(options, "/opt/soft/flink/rocksdb/job_7b558d735e42a1fdf086cd72f97d3803_op_StreamingJoinOperator_72a7ad5de6d319a422288fc6cb9392ae__1_1__uuid_593aa555-c699-4f8f-a3d1-e7e07d90458a/db")) {
//                // 读取指定键的值
//                byte[] value = db.get("myKey".getBytes());
//                System.out.println("Value: " + new String(value));
//
//
//            }
//        } catch (RocksDBException e) {
//            e.printStackTrace();
//        }
    }
}
